package com.market_analysis

import java.sql.Timestamp

import com.market_analysis.bean.{AdClickCountByProvince, AdClickLog}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/12/4 10:43
  *                统计地区的广告点击量
  */
object AdClickAnalysis {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    //读取数据，转换成样例类提取时间戳和watermark
    val filePath: String = "MarketAnalysis/src/main/resources/AdClickLog.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    val dataStream: DataStream[AdClickLog] = fileStream.map(
      data => {
        val arr: Array[String] = data.split(",")
        AdClickLog(arr(0).toLong, arr(1).toLong, arr(2), arr(3), arr(4).toLong)
      }
    )
      .assignAscendingTimestamps(_.timestamp * 1000L)

    //开窗聚合统计
    val resultStream: DataStream[AdClickCountByProvince] = dataStream.keyBy(_.province)
      .timeWindow(Time.days(1), Time.seconds(5))
      .aggregate(new AdCountAgg, new AdCountWindowResult)

    resultStream.print("ad")

    env.execute("ad count by province job")

  }

  class AdCountAgg extends AggregateFunction[AdClickLog, Long, Long] {
    override def createAccumulator(): Long = 0L

    override def add(in: AdClickLog, acc: Long): Long = acc + 1

    override def getResult(acc: Long): Long = acc

    override def merge(acc: Long, acc1: Long): Long = acc + acc1
  }

  class AdCountWindowResult extends WindowFunction[Long,AdClickCountByProvince,String,TimeWindow]{
    override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[AdClickCountByProvince]): Unit = {
      val windowEnd=new Timestamp(window.getEnd).toString

      out.collect(AdClickCountByProvince(windowEnd,key,input.head))
    }
  }

}
